/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.demos;

import com.skjegstad.mist2.IMistSubscriptionListener;
import com.skjegstad.mist2.MistCoreManager;
import com.skjegstad.mist2.datatypes.DataElement;
import com.skjegstad.mist2.datatypes.SimpleEntry;
import com.skjegstad.mist2.exceptions.MistException;
import java.io.BufferedOutputStream;
import java.io.IOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.util.Comparator;
import java.util.Map.Entry;
import java.util.PriorityQueue;
import java.util.concurrent.PriorityBlockingQueue;

/**
 *
 * @author Magnus Skjegstad
 */
public class TestMistOutputPipe {
    public static void main(String[] argv) throws UnknownHostException, InterruptedException, MistException, IOException {
        MistCoreManager mist = new MistCoreManager();
        //mist.enableBroadcast(InetAddress.getByName("10.0.0.255"), 1234);
        mist.enableMulticast(InetAddress.getByName("224.0.0.1"), 1234);
        mist.setDebug(false);

        final PriorityBlockingQueue<Entry<Integer, byte[]>> queue = new PriorityBlockingQueue<Entry<Integer, byte[]>>(10, new Comparator<Entry<Integer, byte[]>>() {
            public int compare(Entry<Integer, byte[]> o1, Entry<Integer, byte[]> o2) {
                return o1.getKey().compareTo(o2.getKey());
            }
        });

        mist.subscribe("no.ffi.stream.test", new IMistSubscriptionListener() {
            public void deliverMessage(DataElement message) {
                try {
                    //System.out.println("in deliver - adding element to queue");
                    ByteBuffer b = ByteBuffer.wrap(message.getPayload());
                    int frag = b.getInt();
                    byte[] data = new byte[message.getPayloadLen() - 4];
                    b.get(data);
                    queue.add(new SimpleEntry<Integer, byte[]>(frag, data));

                   /* String s = "[";
                    for (Entry<Integer, byte[]> q : queue) {
                        s = s + q.getKey() + ",";
                    }
                    s = s + "]";
                    System.err.println("QUEUE: " + s);*/
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        });

        mist.start();
        
        Thread.sleep(5000); // wait 1 sec before starting
        int lastFrag = -1;
        int retries = 0;
        int maxRetries = 3;
        long recvSoFar = 0;
        long start = System.currentTimeMillis();
        while(true) {
            Entry<Integer, byte[]> e = queue.take();

            if (e.getKey() == lastFrag+1 || lastFrag == -1 || e.getKey() > lastFrag + 100) {
                recvSoFar = recvSoFar + e.getValue().length + 4;
                double bw = recvSoFar / ((double)(System.currentTimeMillis() - start) / 1000.0);
                System.err.println("Got frag " + e.getKey() + " (" + (e.getValue().length+4) + " bytes). Elements in queue: " + queue.size() + ", bw=" + bw);
                lastFrag = e.getKey();
                BufferedOutputStream os = new BufferedOutputStream(System.out);
                //OutputStream os = System.out;
                os.write(e.getValue());
                os.flush();
            } else {
                if (e.getKey() > lastFrag) {
                    queue.add(e); // put element back in queue
                   // Thread.sleep(1000); // wait a little
                   /* retries++;
                    if (retries > maxRetries) {
                        System.err.println("Giving up on fragment " + (lastFrag+1) + ". Skipping.");
                        lastFrag = -1;
                        retries = 0;
                    }*/

                }
            }
        }

        //mist.stop();

    }
}
